#nullable disable

using Crux.Core.Websockets.Metadata;
using Crux.Core.Websockets.Redis.Metadata;
using Microsoft.Extensions.Logging;
using Nito.AsyncEx;
using StackExchange.Redis;

namespace Crux.Core.Websockets.Redis;
/// <summary>
/// RedisLifetimeManagerOfT
/// </summary>
public class RedisLifetimeManager<THandler>
    : WebSocketLifetimeManager<THandler>,
      IDisposableObservable
      where THandler : WebSocketHandler
{
    #region private fields
    /// <inheritdoc cref="IConnectionMultiplexer"/>
    private readonly IConnectionMultiplexer _multiplexer;
    /// <inheritdoc cref="ISubscriber"/>
    private readonly ISubscriber _subscriber;
    /// <inheritdoc cref="WebSocketConnectionStore"/>
    private readonly WebSocketConnectionStore _store = new WebSocketConnectionStore();
    /// <inheritdoc cref="ILogger"/>
    private readonly ILogger<RedisLifetimeManager<THandler>> _logger;
    /// <inheritdoc cref="RedisSubChannel"/>
    private readonly RedisSubChannel _redisChannel;
    /// <inheritdoc cref="RedisSubscriptionManager"/>
    private readonly RedisSubscriptionManager _user = new RedisSubscriptionManager();
    /// <inheritdoc cref="RedisSubscriptionManager"/>
    private readonly RedisSubscriptionManager _group = new RedisSubscriptionManager();
    /// <inheritdoc cref="WebSocketOptions"/>
    private readonly WebSocketOptions<THandler> _option;
    /// <inheritdoc cref="AckHandler"/>
    private readonly AckHandler ackHandler = new AckHandler();
    /// <summary>
    ///
    /// </summary>
    private int internalId;

    #endregion  private fields

    #region .ctor
    /// <summary>
    /// 
    /// </summary>
    /// <param name="redisResolver"></param>
    /// <param name="logger"></param>
    /// <param name="webSocketOptions"></param>
    public RedisLifetimeManager(
        IRedisResolver redisResolver,
        ILogger<RedisLifetimeManager<THandler>> logger,
        WebSocketOptions<THandler> webSocketOptions)
    {
        this._option = webSocketOptions
            ?? throw new ArgumentNullException(nameof(webSocketOptions));
        this._logger = logger
            ?? throw new ArgumentNullException(nameof(logger));
        _multiplexer = redisResolver.Resolve();
        _subscriber = _multiplexer.GetSubscriber();
        _redisChannel = new RedisSubChannel(typeof(THandler).FullName!);

        AsyncContext.Run(() =>
        {
            SubscribeToAll();
            SubscribeToGroupManagementChannel();
            SubscribeToAckChannel();
        });
    }
    #endregion .ctor

    #region implements WebSocketLifetimeManager

    /// <inheritdoc cref="ILifetimeManager"/>
    public override Task AddToGroupAsync(
        string connectionId,
        string groupName,
        CancellationToken cancellationToken = default)
    {
        Requires.NotNullOrEmpty(connectionId, nameof(connectionId));
        Requires.NotNullOrEmpty(groupName, nameof(groupName));

        var context = _store[connectionId];

        return context switch
        {
            var x when x is not null => AddGroupAsyncCore(x, groupName),
            _ => SendGroupActionAndWaitForAck(connectionId, groupName, GroupAction.Add)
        };
    }
    /// <inheritdoc cref="ILifetimeManager"/>
    public override Task OnConnectedAsync(WebSocketConnectionContext context)
    {
        _store.Add(context);

        var connectionTask = SubscribeToConnection(context);

        var userTask = context.UserIdentifier switch
        {
            var x when !string.IsNullOrEmpty(x) => SubscribeToUser(context),
            _ => Task.CompletedTask
        };

        return Task.WhenAll(connectionTask, userTask);
    }
    /// <inheritdoc cref="ILifetimeManager"/>
    public override Task OnDisconnectedAsync(WebSocketConnectionContext context)
    {
        _store.Remove(context);

        var connectionChannelName = _redisChannel.Connection(context.ConnectionId);

        var connectionChannel = RedisChannel.Literal(connectionChannelName);

        return Task.WhenAll(RemoveGroupsAsync(context),
                            _subscriber.UnsubscribeAsync(connectionChannel),
                            RemoveUserAsync(context));
    }

    /// <inheritdoc cref="ILifetimeManager"/>
    public override Task RemoveToGroupAsync(
        string connectionId,
        string groupName,
        CancellationToken cancellationToken = default)
    {
        Requires.NotNullOrEmpty(connectionId, nameof(connectionId));
        Requires.NotNullOrEmpty(groupName, nameof(groupName));

        var context = _store[connectionId];
        if (context is not null)
        {
            return RemoveGroupAsyncCore(context, groupName);
        }

        return SendGroupActionAndWaitForAck(connectionId, groupName, GroupAction.Remove);
    }
    /// <inheritdoc cref="ILifetimeManager"/>
    public override Task SendAllAsync(
        string methodName,
        object[] args,
        CancellationToken cancellationToken = default)
    {
        Requires.NotNullOrEmpty(methodName, nameof(methodName));

        var buffer = MessageResolver.SerializeInternalMessage(new InternalMessage
        {
            Method = methodName,
            Arguments = args,
        }, cancellationToken);

        var channelName = _redisChannel.All;

        return PublishAsync(channelName, buffer);
    }
    /// <inheritdoc cref="ILifetimeManager"/>
    public override Task SendAllExceptAsync(
        string methodName,
        object[] args,
        IReadOnlyList<string> exceptList,
        CancellationToken cancellationToken = default)
    {
        Requires.NotNullOrEmpty(methodName, nameof(methodName));

        var buffer = MessageResolver.SerializeInternalMessage(new InternalMessage
        {
            Method = methodName,
            Arguments = args,
            ExcludedConnectionIds = exceptList
        }, cancellationToken);

        var channelName = _redisChannel.All;

        return PublishAsync(channelName, buffer);
    }
    /// <inheritdoc cref="ILifetimeManager"/>
    public override Task SendConnectionAsync(
        string connectionId,
        string methodName,
        object[] args,
        CancellationToken cancellationToken = default)
    {
        Requires.NotNullOrEmpty(connectionId, nameof(connectionId));
        Requires.NotNullOrEmpty(methodName, nameof(methodName));

        var context = _store[connectionId];
        if (context is not null)
        {
            return context.SendAsync(new NotifyMessage
            {
                Type = MessageType.Notify,
                Target = methodName,
                Arguments = args
            });
        }

        var buffer = MessageResolver.SerializeInternalMessage(new InternalMessage
        {
            Method = methodName,
            Arguments = args,
            ConnectionId = connectionId,
        }, cancellationToken);

        var channelName = _redisChannel.Connection(connectionId);

        return PublishAsync(channelName, buffer);
    }
    /// <inheritdoc cref="ILifetimeManager"/>
    public override Task SendConnectionsAsync(
        IReadOnlyList<string> connectionIds,
        string methodName,
        object[] args,
        CancellationToken cancellationToken = default)
    {
        Requires.NotNull(connectionIds, nameof(connectionIds));
        Requires.NotNullOrEmpty(methodName, nameof(methodName));

        var option = new ParallelOptions
        {
            CancellationToken = cancellationToken,
            TaskScheduler = TaskScheduler.Current
        };

        var parallelLoopResult = Parallel.ForEach(connectionIds, option, (connectionId, state) =>
        {
            if (state.IsExceptional ||
                state.IsStopped ||
                state.ShouldExitCurrentIteration)
            {
                return;
            }

            try
            {
                AsyncContext.Run(async () => await SendConnectionAsync(connectionId, methodName, args, cancellationToken).ConfigureAwait(false));
                if (cancellationToken.IsCancellationRequested)
                {
                    state.Break();
                }
            }
            catch (Exception e)
            {
                _logger.LogError(e, e.Message);
            }
        });

        return Task.CompletedTask;
    }
    /// <inheritdoc cref="ILifetimeManager"/>
    public override Task SendGroupAsync(
        string groupName,
        string methodName,
        object[] args,
        CancellationToken cancellationToken = default)
    {
        Requires.NotNullOrEmpty(groupName, nameof(groupName));
        Requires.NotNullOrEmpty(methodName, nameof(methodName));

        var buffer = MessageResolver.SerializeInternalMessage(new InternalMessage
        {
            Method = methodName,
            Arguments = args,
            GroupName = groupName
        }, cancellationToken);

        var channelName = _redisChannel.Group(groupName);

        return PublishAsync(channelName, buffer);
    }
    /// <inheritdoc cref="ILifetimeManager"/>
    public override Task SendGroupExceptAsync(
        string groupName,
        string methodName,
        object[] args,
        IReadOnlyList<string> connectionIds,
        CancellationToken cancellationToken = default)
    {
        Requires.NotNullOrEmpty(groupName, nameof(groupName));
        Requires.NotNullOrEmpty(methodName, nameof(methodName));

        var buffer = MessageResolver.SerializeInternalMessage(new InternalMessage
        {
            Method = methodName,
            Arguments = args,
            GroupName = groupName,
            ExcludedConnectionIds = connectionIds
        }, cancellationToken);

        var channelName = _redisChannel.Group(groupName);

        return PublishAsync(channelName, buffer);
    }
    /// <inheritdoc cref="ILifetimeManager"/>
    public override Task SendGroupsAsync(
        IReadOnlyList<string> groupNames,
        string methodName,
        object[] args,
        CancellationToken cancellationToken = default)
    {
        Requires.NotNull(groupNames, nameof(groupNames));
        Requires.NotNullOrEmpty(methodName, nameof(methodName));

        var option = new ParallelOptions
        {
            CancellationToken = cancellationToken,
            TaskScheduler = TaskScheduler.Current
        };

        var parallelLoopResult = Parallel.ForEach(groupNames, option, (groupName, state) =>
        {
            if (state.IsExceptional ||
                state.IsStopped ||
                state.ShouldExitCurrentIteration)
            {
                return;
            }

            try
            {
                AsyncContext.Run(async () => await SendGroupAsync(groupName, methodName, args, cancellationToken).ConfigureAwait(false));
                if (cancellationToken.IsCancellationRequested)
                {
                    state.Break();
                }
            }
            catch (Exception e)
            {
                _logger.LogError(e, e.Message);
            }
        });

        return Task.CompletedTask;
    }
    /// <inheritdoc cref="ILifetimeManager"/>
    public override Task SendUserAsync(
        string userId,
        string methodName,
        object[] args,
        CancellationToken cancellationToken = default)
    {
        Requires.NotNullOrEmpty(userId, nameof(userId));
        Requires.NotNullOrEmpty(methodName, nameof(methodName));

        var buffer = MessageResolver.SerializeInternalMessage(new InternalMessage
        {
            Method = methodName,
            Arguments = args,
            UserId = userId
        }, cancellationToken);

        return PublishAsync(_redisChannel.User(userId), buffer);
    }
    /// <inheritdoc cref="ILifetimeManager"/>
    public override Task SendUsersAsync(
        IReadOnlyList<string> userIds,
        string methodName,
        object[] args,
        CancellationToken cancellationToken = default)
    {
        Requires.NotNull(userIds, nameof(userIds));
        Requires.NotNullOrEmpty(methodName, nameof(methodName));

        var option = new ParallelOptions
        {
            CancellationToken = cancellationToken,
            TaskScheduler = TaskScheduler.Current
        };

        var parallelLoopResult = Parallel.ForEach(userIds, option, (userId, state) =>
        {
            if (state.IsExceptional ||
                state.IsStopped ||
                state.ShouldExitCurrentIteration)
            {
                return;
            }

            try
            {
                AsyncContext.Run(async () => await SendUserAsync(userId, methodName, args, cancellationToken).ConfigureAwait(false));

                if (cancellationToken.IsCancellationRequested)
                {
                    state.Break();
                }
            }
            catch (Exception e)
            {
                _logger.LogError(e, e.Message);
            }
        });

        return Task.CompletedTask;
    }
    #endregion implements WebSocketLifetimeManager
    #region implements IDisposableObservable
    /// <inheritdoc cref="IDisposableObservable"/>
    public void Dispose()
    {
        if (IsDisposed) return;

        IsDisposed = true;
        GC.SuppressFinalize(this);
    }
    /// <inheritdoc cref="IDisposableObservable"/>
    public bool IsDisposed { get; set; }
    #endregion implements IDisposableObservable

    #region private methods
    private void SubscribeToAll()
    {
        var channel = RedisChannel.Literal(_redisChannel.All);

        Action<RedisChannel, RedisValue> handle = (channel, message) =>
        {
            try
            {
                _logger.LogDebug($"receiver message:", message);

                var option = new ParallelOptions
                {
                    TaskScheduler = TaskScheduler.Current
                };

                var contexts = Contexts(_store);

                var parallelLoopResult = Parallel.ForEach(contexts, option, (context, state) =>
                {
                    if (state.IsExceptional ||
                        state.IsStopped ||
                        state.ShouldExitCurrentIteration)
                    {
                        return;
                    }

                    try
                    {
                        AsyncContext.Run(async () => await InternalSendAsync(message, context).ConfigureAwait(false));
                    }
                    catch (Exception e)
                    {
                        _logger.LogError(e, e.Message);
                    }
                });
            }
            catch (Exception e)
            {
                _logger?.LogError(e, e.Message);
            }

            static IEnumerable<WebSocketConnectionContext> Contexts(WebSocketConnectionStore store)
            {
                foreach (var context in store)
                {
                    yield return context;
                }
            }
        };

        _subscriber.Subscribe(channel, handle);
    }

    private Task InternalSendAsync(RedisValue message, WebSocketConnectionContext context)
    {
        var internalMessage = MessageResolver.DeserializeInternalMessage(message, context.ConnectionAborted);

        if (internalMessage.ExcludedConnectionIds is not null &&
            internalMessage.ExcludedConnectionIds.Contains(context.ConnectionId))
        {
            return Task.CompletedTask;
        }

        return context.SendAsync(new NotifyMessage
        {
            Type = MessageType.Notify,
            Target = internalMessage.Method,
            Arguments = internalMessage.Arguments
        });
    }
    private void SubscribeToGroupManagementChannel()
    {
        var channel = RedisChannel.Literal(_redisChannel.GroupManagement);

        Action<RedisChannel, RedisValue> handle = async (channel, message) =>
        {
            try
            {
                var command = MessageResolver.DeserializeGroupCommand(message);
                if (command is null)
                {
                    _logger?.LogDebug("group command is null");
                    return;
                }

                var context = _store[command.ConnectionId];
                if (context is null)
                {
                    _logger?.LogDebug($"context is not found,id:{command.ConnectionId}");
                    return;
                }

                switch (command.Action)
                {
                    case GroupAction.Remove:
                        await RemoveGroupAsyncCore(context, command.GroupName).ConfigureAwait(false);
                        break;

                    case GroupAction.Add:
                        await AddGroupAsyncCore(context, command.GroupName).ConfigureAwait(false);
                        break;
                }

                await PublishAsync(_redisChannel.Ack(command.ServerName), message).ConfigureAwait(false);
            }
            catch (Exception e)
            {
                _logger?.LogError(e, e.Message);
            }
        };

        _subscriber.Subscribe(channel, handle);
    }
    private async Task SendGroupActionAndWaitForAck(
        string connectionId,
        string groupName,
        GroupAction groupAction = default)
    {
        var id = Interlocked.Increment(ref internalId);

        var ack = ackHandler.CreateAck(id);

        var command = new GroupCommand(id, _option.ServerName, groupAction, groupName, connectionId);

        var buffer = MessageResolver.SerializeGroupCommand(command);

        var channelName = _redisChannel.GroupManagement;

        await PublishAsync(channelName, buffer).ConfigureAwait(false);

        await ack;
    }
    private void SubscribeToAckChannel()
    {
        var channelName = _redisChannel.Ack(_option.ServerName);

        var channel = RedisChannel.Literal(channelName);

        Action<RedisChannel, RedisValue> handle = (channel, message) =>
        {
            var command = MessageResolver.DeserializeGroupCommand(message);
            if (command is null)
            {
                return;
            }

            ackHandler.TriggerAck(command.Id);
        };

        _subscriber.Subscribe(channel, handle);
    }
    private async Task SubscribeToConnection(WebSocketConnectionContext context)
    {
        var listenChannel = _redisChannel.Connection(context.ConnectionId);

        var channel = RedisChannel.Literal(listenChannel);

        Action<RedisChannel, RedisValue> handle = async (channel, message) =>
        {
            try
            {
                await InternalSendAsync(message, context).ConfigureAwait(false);
            }
            catch (Exception e)
            {
                _logger?.LogError(e, e.Message);
            }
        };

        await _subscriber.SubscribeAsync(channel, handle).ConfigureAwait(false);
    }
    private Task SubscribeToUser(WebSocketConnectionContext context)
    {
        var userChannel = _redisChannel.User(context.UserIdentifier);

        Func<string, WebSocketConnectionStore, Task> subscribeMethod = async (channelName, subscriptions) =>
        {
            var channel = RedisChannel.Literal(channelName);

            Action<RedisChannel, RedisValue> handle = async (channel, message) =>
            {
                try
                {
                    await InternalSendAsync(message, context).ConfigureAwait(false);
                }
                catch (Exception e)
                {
                    _logger?.LogError(e, e.Message);
                }
            };

            await _subscriber.SubscribeAsync(channel, handle).ConfigureAwait(false);
        };

        return _user.AddSubscriptionAsync(userChannel, context, subscribeMethod);
    }
    private async Task RemoveGroupsAsync(WebSocketConnectionContext context)
    {
        var cloneGroup = context.InternalGroups.ToArray();

        foreach (var groupName in cloneGroup)
        {
            await RemoveGroupAsyncCore(context, groupName).ConfigureAwait(false);
        }
    }
    private async Task RemoveGroupAsyncCore(
        WebSocketConnectionContext context,
        string groupName)
    {
        Requires.NotNullOrEmpty(groupName, nameof(groupName));

        var groupChannel = _redisChannel.Group(groupName);

        Func<string, Task> unsubscribeMethod = async channelName =>
        {
            var channel = RedisChannel.Literal(channelName);

            await _subscriber.UnsubscribeAsync(channel).ConfigureAwait(false);
        };

        await _group.RemoveSubscriptionAsync(groupChannel, context, unsubscribeMethod).ConfigureAwait(false);

        context.InternalGroups.RemoveWhere(p => p == groupName);
    }
    private async Task AddGroupAsyncCore(
        WebSocketConnectionContext context,
        string groupName)
    {
        Requires.NotNullOrEmpty(groupName, nameof(groupName));

        if (!context.InternalGroups.Add(groupName))
        {
            return;
        }

        var groupChannel = _redisChannel.Group(groupName);

        await _group.AddSubscriptionAsync(groupChannel, context, SubscribeToGroupAsync(context)).ConfigureAwait(false);
    }
    private Func<string, WebSocketConnectionStore, Task> SubscribeToGroupAsync(WebSocketConnectionContext context)
    {
        return async (groupChannel, subscription) =>
        {
            var channel = RedisChannel.Literal(groupChannel);

            Action<RedisChannel, RedisValue> handle = async (channel, message) =>
            {
                try
                {
                    await InternalSendAsync(message, context).ConfigureAwait(false);
                }
                catch (Exception e)
                {
                    _logger?.LogError(e, e.Message);
                }
            };

            await _subscriber.SubscribeAsync(channel, handle).ConfigureAwait(false);
        };
    }
    private Task RemoveUserAsync(WebSocketConnectionContext context)
    {
        if (string.IsNullOrEmpty(context.UserIdentifier))
        {
            return Task.CompletedTask;
        }

        var userChannel = _redisChannel.User(context.UserIdentifier);

        Func<string, Task> unsubscribeMethod = async channelName =>
        {
            var channel = RedisChannel.Literal(channelName);

            await _subscriber.UnsubscribeAsync(channel).ConfigureAwait(false);
        };

        return _user.RemoveSubscriptionAsync(userChannel, context, unsubscribeMethod);
    }
    private async Task PublishAsync(
        string channelName,
        byte[] payload)
    {
        var channel = RedisChannel.Literal(channelName);

        _ = await _subscriber.PublishAsync(channel, payload).ConfigureAwait(false);
    }
    #endregion private methods
}